package com.poetic.network.flow.process;

import com.poetic.network.flow.bloom.Bloom;
import com.poetic.network.flow.domain.UvCount;
import lombok.val;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

/**
 * <pre>
 *  TODO
 * Created by lianghuikun on 2020-09-15.
 * </pre>
 *
 * @author lianghuikun
 */
public class UVCountWithBloom extends ProcessWindowFunction<Tuple2<String, Long>, UvCount, Long, TimeWindow> {

    //   // 定义redis连接
    private Jedis jedis = new Jedis("localhost", 6379);
    private Bloom bloom = new Bloom(1 << 29);

    @Override
    public void process(Long aLong, Context context, Iterable<Tuple2<String, Long>> elements, Collector<UvCount> out) throws Exception {
        // 位图的存储方式，key是windowEnd，value是bitmap
        String storeKey = context.window().getEnd() + "";
        Integer count = 0;
        // 把每个窗口的uv count值也存入名为count的redis表，存放内容为（windowEnd -> uvCount），所以要先从redis中读取
        if (jedis.hget("count", storeKey) != null) {
            count = Integer.valueOf(jedis.hget("count", storeKey));
        }
        // 用布隆过滤器判断当前用户是否已经存在

        // elements中只有一条数据
        Long userId = elements.iterator().next().f1;
        Long offset = bloom.hash(userId.toString(), 61);
        // 定义一个标识位，判断reids位图中有没有这一位
        Boolean isExist = jedis.getbit(storeKey, offset);
        if (isExist) {
            UvCount uvCount = UvCount.builder()
                    .windowEnd(Long.valueOf(storeKey))
                    .count(count)
                    .build();
            out.collect(uvCount);
        } else {
            // 如果不存在，位图对应位置1，count + 1
            jedis.setbit(storeKey, offset, true);
            jedis.hset("count", storeKey, (count + 1) + "");
            UvCount uvCount = UvCount.builder()
                    .windowEnd(Long.valueOf(storeKey))
                    .count(count + 1)
                    .build();
            out.collect(uvCount);

        }
    }
}
